#ifndef SDFS_SERVER_H
#define SDFS_SERVER_H

#include <muduo/base/Atomic.h>
#include <muduo/base/Logging.h>
#include <muduo/base/Thread.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/InetAddress.h>
#include <muduo/net/TcpClient.h>

#include <boost/bind.hpp>

#include <sdfs/protobuf/Codec.h>
#include <sdfs/protobuf/Message.h>
#include <sdfs/protobuf/ProtobufDispatcher.h>
#include <sdfs/config/StorageConfig.h>
#include <sdfs/storage/FileService.h>
#include <sdfs/storage/Job.h>

using namespace muduo;
using namespace muduo::net;
using namespace sdfs::protobuf;

namespace sdfs
{
namespace storage
{

class StorageServer
{
public:
	StorageServer(EventLoop* loop, const InetAddress& addr, const InetAddress& trackerAddr, int threadNum)
		: trackerAddr_(trackerAddr),
		 server_(loop, addr, "STORAGE_SERVER"),
		 client_(new TcpClient(loop, trackerAddr_, "TRACKER_CLIENT")), 
		 dispatcher_( 
		 	boost::bind(&StorageServer::discardMessage, this, _1, _2, _3)
		 ),
		 codec_(
		 	boost::bind(&ProtobufDispatcher::onProtobufMessage, &dispatcher_, _1, _2, _3)
		 ),
	 	 fileService_(&(*loop), client_->connection(), &codec_, 
	 	 	sdfs::StorageConfig::blockReaderNum()), 
		 lunchTime_(Timestamp::now()),
		 threadNum_(threadNum)
	{
		server_.setConnectionCallback(
			boost::bind(&StorageServer::onConnection, this, _1));
		server_.setMessageCallback(
			boost::bind(&ProtobufCodec::onMessage, &codec_, _1, _2, _3));

		client_->setConnectionCallback(
			boost::bind(&StorageServer::onConnectTracker, this, _1));
		client_->setMessageCallback(
			boost::bind(&ProtobufCodec::onMessage, &codec_, _1, _2, _3));
/*
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::SaveRequest>(
			boost::bind(&StorageServer::onSave, this, _1, _2, _3));
*/
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::Keepalive>(
			boost::bind(&StorageServer::onHeartBeat, this, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::Sort>(
			boost::bind(&StorageServer::onSort, this, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::MergeMessage>(
			boost::bind(&StorageServer::onMerge, this, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileSaveRequestMessage>(
			boost::bind(&FileService::onFileSaveRequest, &fileService_, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileSyncMessage>(
			boost::bind(&FileService::onAddSyncTask, &fileService_, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileFindResultMessage>(
			boost::bind(&FileService::onSyncFile, &fileService_, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileDownloadRequestMessage>(
			boost::bind(&FileService::onDownloadFile, &fileService_, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileDownloadResultMessage>(
			boost::bind(&FileService::onRecieveFile, &fileService_, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileReadyMessage>(
			boost::bind(&FileService::onSendFile, &fileService_, _1, _2, _3));
	}
	//onSort
	void onSort(const muduo::net::TcpConnectionPtr& conn,
		const shared_ptr<sdfs::protobuf::Sort>& message, muduo::Timestamp timestamp)
	{	
		LOG_DEBUG << "Sort Task, size: " << message->size() << " direction: " ;
		std::vector<int> input;
		for (int i = 0; i < message->size(); ++i)
		{
			input.push_back(message->data(i));
		}
		Jobs::Sort<int>(input);
		sdfs::protobuf::Sort respMsg;
		respMsg.set_size(input.size());
		for (uint i = 0; i < input.size(); ++i)
		{
			respMsg.add_data(input.at(i));
		}
		LOG_DEBUG << "send message back, size: " << respMsg.size();

	}
	//onMerge
	void onMerge(const muduo::net::TcpConnectionPtr& conn,
		const shared_ptr<sdfs::protobuf::MergeMessage>& message, muduo::Timestamp timestamp)
	{
		LOG_DEBUG << "Merge Task, size1: " << message->size1() << " size2: " << message->size2();

		std::vector<int> *input1 = new std::vector<int>();
		std::vector<int> *input2= new std::vector<int>();
		std::vector<int> *result= new std::vector<int>();
		for (int i = 0; i < message->size1(); ++i)
		{
			/* code */
			input1->push_back(message->data1(i));
		}
		for (int i = 0; i < message->size2(); ++i)
		{
			/* code */
			input2->push_back(message->data2(i));
		}


		Jobs::Merge<int>(*result, *input1, *input2);
		sdfs::protobuf::Sort respMsg;
		respMsg.set_size(result->size());
		for (uint i = 0; i < result->size(); ++i)
		{
			/* code */
			respMsg.add_data(result->at(i));
		}
		LOG_DEBUG << "send message back, size: " << respMsg.size();

		codec_.Send(conn, respMsg);
		conn->shutdown();
	}

	void onConnectTracker(const TcpConnectionPtr& conn)
	{
		LOG_TRACE << "[" << conn->localAddress().toIpPort()
			<< "] -> [" << conn->peerAddress().toIpPort() << "]" <<
			" is " << (conn->connected() ? "UP" : "DOWN");
		if(conn->connected())
		{
			LOG_DEBUG << "send HeartBeat to Tracker";

			fileService_.setTrackerConnection(conn);

			sdfs::protobuf::HeartBeat message;
			message.set_id(msgIndex_.addAndGet(1));
			message.set_name(StorageConfig::name().c_str());
			message.set_free(100);
			message.set_groupidx(StorageConfig::groupIdx());
			message.set_capacity(StorageConfig::storageCapacityMB());
			message.set_lunchtime(lunchTime_.microSecondsSinceEpoch());
			message.set_type(sdfs::protobuf::HeartBeat::FIRST);
			message.set_lastsynctime(fileService_.lastSyncTime());
			message.set_ip(StorageConfig::ip().c_str());
			message.set_port(StorageConfig::filePort());
			codec_.Send(conn, message);
		}
		else
		{
			LOG_ERROR << "Tracker[" << client_->connection()->peerAddress().toIpPort()
				<< "] had lost! retrying...";
			// client_->connection()->connectDestroyed();
			fileService_.resetTrackerConnection();
			server_.getLoop()->runAfter(1.0, 
		  		boost::bind(&StorageServer::reconnectTracker, this));
		}
	}
	
	void reconnectTracker()
	{
		client_.reset(new TcpClient(server_.getLoop(), trackerAddr_, "TRACKER_CLIENT"));
		client_->setConnectionCallback(
			boost::bind(&StorageServer::onConnectTracker, this, _1));
		client_->setMessageCallback(
			boost::bind(&ProtobufCodec::onMessage, &codec_, _1, _2, _3));
		client_->connect();
	}


	void discardMessage(const TcpConnectionPtr& conn, 
		const MessagePtr& message, Timestamp timestamp)
	{
		LOG_TRACE << "message["<<message->GetTypeName()
			<< "] from: " << conn->peerAddress().toIpPort()
			<< " discard";
	}

	void onConnection(const TcpConnectionPtr& conn)
	{
		LOG_DEBUG << conn->peerAddress().toIpPort() <<
			" -> " << conn->localAddress().toIpPort() <<
			" is " << (conn->connected() ? "UP" : "DOWN");
		UploadContextPtr context = boost::any_cast<UploadContextPtr>(conn->getContext() );
		if(context != NULL && !context->piece->isFinished())
		{
			
		}

	}
	void onHeartBeat(const TcpConnectionPtr& conn, 
		const KeepalivePtr& keepalive, Timestamp timestamp)
	{
		LOG_TRACE << "keepalive from: " << conn->peerAddress().toIpPort();
		sdfs::protobuf::HeartBeat message;
		message.set_id(msgIndex_.addAndGet(1));
		message.set_groupidx(StorageConfig::groupIdx());
		message.set_capacity(StorageConfig::storageCapacityMB());
		message.set_free(1024);
		message.set_lunchtime(lunchTime_.microSecondsSinceEpoch());
		message.set_name(StorageConfig::name().c_str());
		message.set_type(sdfs::protobuf::HeartBeat::NORMAL);
		message.set_lastsynctime(fileService_.lastSyncTime());

		codec_.Send(conn, message);
	}

	

	void start()
	{
		LOG_INFO << "starting storage server";
		server_.start();

		client_->connect();
/*
		client_->getLoop()->runEvery(
			Configuration::getHeartBeatTimeIntervalMs()/1000.0, \
			boost::bind(&StorageServer::onHeartBeat, this));
*/
	}

	
	

	/* data */
private:
	// EventLoop *loop_;
	InetAddress trackerAddr_;
	TcpServer server_;
	boost::shared_ptr<TcpClient> client_;
	ProtobufDispatcher dispatcher_;
	ProtobufCodec codec_;
	FileService fileService_;
	Timestamp lunchTime_;
	int threadNum_;
	AtomicInt32 msgIndex_;
	
};

}
}
#endif
